package org.intrasearch.spider;

import org.intrasearch.core.ArtifactId;
import org.intrasearch.core.ArtifactStorage;
import org.intrasearch.core.InvertedIndex;
import org.intrasearch.core.ResourceArtifact;
import org.intrasearch.core.mapreduce.MapReduce;
import org.intrasearch.core.mapreduce.MapTask;

import java.util.StringTokenizer;

public class SpiderMapTask implements MapTask<SpiderMapTaskInput, String, ArtifactId>
{
    private final StatusLogger logger;
    private final ArtifactStorage artifactStorage;

    public SpiderMapTask(StatusLogger logger, ArtifactStorage artifactStorage)
    {
        this.logger = logger;
        this.artifactStorage = artifactStorage;
    }

    public void map(SpiderMapTaskInput input, MapReduce<SpiderMapTaskInput, String, ArtifactId> virtualFileStringArtifactIdMapReduce)
    {
        startIndexingRecursive(input, virtualFileStringArtifactIdMapReduce);
    }

    private void startIndexingRecursive(SpiderMapTaskInput mapInput, MapReduce<SpiderMapTaskInput, String, ArtifactId> mapReduce)
    {
        logger.logStatus("Indexing dir: " + mapInput.dir.toString());
        ArtifactId dirArtifactId = indexFile(mapInput.dir, mapInput.parentDirId, mapReduce);
        VirtualFile[] files = null;
        try
        {
            files = mapInput.dir.listFiles();
        }
        catch (Exception e)
        {
            logger.logError("Error while listing " + mapInput, e);
        }
        if (files == null)
        {
            return;
        }
        for (VirtualFile file : files)
        {
            if (file.isFile())
            {
                indexFile(file, mapInput.parentDirId, mapReduce);
            }
            else if (file.isDirectory())
            {
                startIndexingRecursive(new SpiderMapTaskInput(file, dirArtifactId), mapReduce);
            }
        }
    }

    private ArtifactId indexFile(VirtualFile file, ArtifactId parentDirId, MapReduce<SpiderMapTaskInput, String, ArtifactId> mapReduce)
    {
        try
        {
            ResourceArtifact<? extends ResourceArtifact> artifact = Artifacts.createFromVirtualFile(file, parentDirId);
            artifactStorage.addArtifact(artifact);
            mapArtifact(mapReduce, artifact);
            return artifact.getArtifactId();
        }
        catch (Exception e)
        {
            logger.logError("Error while indexing " + file, e);
        }
        return null;
    }

    private void mapArtifact(MapReduce<SpiderMapTaskInput, String, ArtifactId> mapReduce, ResourceArtifact<? extends ResourceArtifact> fileArtifact)
    {
        for (StringTokenizer stringTokenizer = new StringTokenizer(fileArtifact.getName(), InvertedIndex.WORD_DELIMETER); stringTokenizer.hasMoreTokens();)
        {
            String token = stringTokenizer.nextToken();
            mapReduce.map(token, fileArtifact.getArtifactId());
        }
    }
}
